package com.atguigu.day09;

import com.atguigu.day04.Example10;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

//实时热门商品
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> streamSource = env.readTextFile("D:\\WebSite\\flink\\flink\\java_flink\\src\\main\\resources\\UserBehavior.csv");
        SingleOutputStreamOperator<Example10.UserBehavior> stream = streamSource
                .map(x -> {
                    String[] arr = x.split(",");
                    return new Example10.UserBehavior(
                            arr[0],
                            arr[1],
                            arr[2],
                            arr[3],
                            Long.parseLong(arr[4]));
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Example10.UserBehavior>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Example10.UserBehavior>() {
                                    @Override
                                    public long extractTimestamp(Example10.UserBehavior element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                }))
                .filter(x -> x.behavior.equals("pv"));

        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env, settings);
        Table table = streamTableEnvironment.fromDataStream(
                stream,
                $("itemId").as("itemId"),
                $("timestamp").rowtime().as("ts")
        );
        streamTableEnvironment.createTemporaryView("userBehavior",table);
        String innerSQL = "select itemId ," +
                " COUNT(itemId) as cnt," +
                "HOP_START(ts,INTERVAL '5' MINUTES ,INTERVAL '1' HOURS) as windowStart," +
                "HOP_END(ts,INTERVAL '5' MINUTES ,INTERVAL '1' HOURS) as windowEnd " +
                "FROM userBehavior GROUP BY itemId , HOP(ts,INTERVAL '5' MINUTES ,INTERVAL '1' HOURS)";
        String overSQL ="select * ,ROW_NUMBER() OVER(partition by windowEnd order by cnt desc) as row_num from ("+innerSQL+")";
        String finalSQL ="select * from ("+overSQL+") where row_num <= 3";
        Table result = streamTableEnvironment.sqlQuery(finalSQL);
        streamTableEnvironment.toChangelogStream(result).print();

        env.execute();
    }
}
